package com.chenwc.rocketmq.example;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * RocketMQ事务消息是一种消息传递机制，旨在保证消息发送与本地事务处理的原子性，即它们要么同时成功，要么同时失败。
 * 这种机制在分布式系统中非常有用，尤其是在需要确保数据一致性的场景下。
 * 工作流程
 * 1、发送半消息：生产者向RocketMQ发送一条半消息（Half Message），这条消息此时处于“待确认”状态，不会被消费者消费。半消息包含了一个唯一的transactionId，用于后续的状态查询。
 * 2、执行本地事务：生产者在发送半消息后，开始执行本地事务逻辑。本地事务可以是任何需要与消息发送同步的操作，如数据库更新等。
 * 3、提交事务状态：在本地事务执行完成后，生产者根据事务的执行结果向RocketMQ提交事务状态（COMMIT或ROLLBACK）。
 * 如果生产者提交COMMIT状态，RocketMQ将半消息标记为可消费状态，并将其投递给消费者。
 * 如果生产者提交ROLLBACK状态，RocketMQ将直接删除半消息，消费者不会收到该消息。
 * 4、事务回查：如果生产者在一段时间内没有提交事务状态，RocketMQ会主动发起回查请求（Check Transaction Status），询问生产者的事务状态。
 *
 * @author chenwc
 */
public class ProducerTransactionMessageExample {
    private static final Logger logger = LoggerFactory.getLogger(ProducerTransactionMessageExample.class);

    private ProducerTransactionMessageExample() {
    }

    public static void main(String[] args) throws MQClientException {
        // 定义Topic
        String topic = "TopicTest";

        // 实现TransactionListener接口
        TransactionListener transactionListener = new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                // 2、执行本地事务：生产者在发送半消息后，开始执行本地事务逻辑。本地事务可以是任何需要与消息发送同步的操作，如数据库更新等。
                logger.info("模拟执行本地事务：生产者在发送半消息后，开始执行本地事务逻辑。本地事务可以是任何需要与消息发送同步的操作，如数据库更新等");

//                try {
//                    Thread.sleep(60000);
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
                logger.info("模拟执行本地事务成功");

                // 3、提交事务状态：在本地事务执行完成后，生产者根据事务的执行结果向RocketMQ提交事务状态（COMMIT或ROLLBACK）。
                // 如果生产者提交COMMIT状态，RocketMQ将半消息标记为可消费状态，并将其投递给消费者。
                // 如果生产者提交ROLLBACK状态，RocketMQ将直接删除半消息，消费者不会收到该消息。
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                // 如果生产者在一段时间内没有提交事务状态，RocketMQ会主动发起回查请求（Check Transaction Status），询问生产者的事务状态。
                logger.info("如果生产者在一段时间内没有提交事务状态，RocketMQ会主动发起回查请求（Check Transaction Status），询问生产者的事务状态");
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        };
        final TransactionMQProducer producer = ProducerSingleton.getTransactionInstance(transactionListener, topic);

        // 定义消息体
        byte[] body = "This is a transaction message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
        // 定义消息标签
        String tag = "TagA";
        // 定义消息key
        String keys = UUID.randomUUID().toString().toUpperCase().replace("-", "");
        // 创建消息对象
        final Message message = new Message(topic, tag, keys, body);

        try {
            // 1、发送半消息：生产者向RocketMQ发送一条半消息（Half Message）
            final TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, null);
            logger.info("Send transaction message successfully, messageId={}", transactionSendResult.toString());
        } catch (Throwable t) {
            logger.error("Failed to send message", t);
        }
        // 关闭生产者
        ProducerSingleton.closeTransactionInstance();
    }

    /**
     * 实现TransactionListener接口
     */
    static class TransactionListenerImpl implements TransactionListener {

        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            // 2、执行本地事务：生产者在发送半消息后，开始执行本地事务逻辑。本地事务可以是任何需要与消息发送同步的操作，如数据库更新等。
            logger.info("模拟执行本地事务：生产者在发送半消息后，开始执行本地事务逻辑。本地事务可以是任何需要与消息发送同步的操作，如数据库更新等");

            // 3、提交事务状态：在本地事务执行完成后，生产者根据事务的执行结果向RocketMQ提交事务状态（COMMIT或ROLLBACK）。
            // 如果生产者提交COMMIT状态，RocketMQ将半消息标记为可消费状态，并将其投递给消费者。
            // 如果生产者提交ROLLBACK状态，RocketMQ将直接删除半消息，消费者不会收到该消息。
            return LocalTransactionState.COMMIT_MESSAGE;
            // return LocalTransactionState.ROLLBACK_MESSAGE;
            // return LocalTransactionState.UNKNOW;
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            // 如果生产者在一段时间内没有提交事务状态，RocketMQ会主动发起回查请求（Check Transaction Status），询问生产者的事务状态。生产者需要实现一个TransactionCheckListener接口，用来处理Broker发起的回查请求，并返回当前事务的状态。
            logger.info("如果生产者在一段时间内没有提交事务状态，RocketMQ会主动发起回查请求（Check Transaction Status），询问生产者的事务状态。生产者需要实现一个TransactionCheckListener接口，用来处理Broker发起的回查请求，并返回当前事务的状态");
            return LocalTransactionState.COMMIT_MESSAGE;
            // return LocalTransactionState.ROLLBACK_MESSAGE;
            // return LocalTransactionState.UNKNOW;
        }
    }
}
